package com.cn.zengzeng.spbzengzeng.controller.queue;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cn.zengzeng.spbzengzeng.utils.KafkaUtils;
import com.cn.zengzeng.spbzengzeng.utils.ThreadPoolExecutorFactory;
import com.cn.zengzeng.spbzengzeng.vo.ActionResult;
import com.google.common.collect.Lists;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.*;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * kafka控制器
 *
 * @author 154594742@qq.com
 * @date 2021/3/2 15:01
 */

@RestController
@Slf4j
@Api(value = "kafka相关信息操作")
public class KafkaController {

    @Autowired
    private KafkaUtils kafkaUtils;
    private final Executor threadPoolUtils = ThreadPoolExecutorFactory.get(1000);

    /**
     * 新增topic (支持批量，这里就单个作为演示)
     *
     * @param topic topic
     * @return ResponseVo
     */
    @PostMapping("kafka")
    @ApiOperation(value="新增topic", notes="")
    @ApiImplicitParam(paramType="body", name = "topic", value = "topic名称", required = true, dataType = "String",dataTypeClass=String.class)
    public ActionResult add(@RequestParam String topic) {
        NewTopic newTopic = new NewTopic(topic, 3, (short) 1);
        kafkaUtils.createTopic(Lists.newArrayList(newTopic));
        return new ActionResult("200","创建成功","");
    }

    /**
     * 查询topic信息 (支持批量，这里就单个作为演示)
     *
     * @param topic 自增主键
     * @return ResponseVo
     */
    @ApiOperation(value="查询topic", notes="")
    @GetMapping("kafka/{topic}")
    public ActionResult getBytTopic(@PathVariable("topic") String topic) {
        return new ActionResult("200","查询成功",kafkaUtils.getTopicInfo(Lists.newArrayList(topic)));
    }

    /**
     * 删除topic (支持批量，这里就单个作为演示)
     * (注意：如果topic正在被监听会给人感觉删除不掉（但其实是删除掉后又会被创建）)
     *
     * @param topic topic
     * @return ResponseVo
     */
    @ApiOperation(value="删除topic", notes="")
    @DeleteMapping("kafka/{topic}")
    public ActionResult delete(@PathVariable("topic") String topic) {
        kafkaUtils.deleteTopic(Lists.newArrayList(topic));
        return new ActionResult("200","删除成功","");
    }

    /**
     * 查询所有topic
     *
     * @return ResponseVo
     */
    @ApiOperation(value="查询所有topic", notes="")
    @GetMapping("kafka/allTopic")
    public ActionResult getAllTopic() {
        return new ActionResult("200","查询成功",kafkaUtils.getAllTopic());
    }

    /**
     * 生产者往topic中发送消息demo
     *
     * @param topic
     * @param message
     * @return
     */
    @ApiOperation(value="往topic中发送消息", notes="")
    @PostMapping("kafka/message")
    public ActionResult sendMessage(@RequestParam(value = "topic") String topic,
                                    @RequestParam(value = "message") String message) {
        kafkaUtils.sendMessage(topic, message);
        return new ActionResult("200","发送成功","");
    }

    /**
     * 消费者示例demo
     * <p>
     * 基于注解监听多个topic，消费topic中消息
     * （注意：如果监听的topic不存在则会自动创建）
     */
    @KafkaListener(topics = {"topic1", "topic2", "topic3"})
    public void consume(String message) {
        log.info("receive msg: " + message);
    }
    @ApiOperation(value="foreach测试线程池", notes="")
    @PostMapping(value = "/insertorderoms")
    public void insertOrderOms(@RequestBody JSONObject jsonObject){
        JSONArray orders = jsonObject.getJSONArray("orders");
        log.info("receive_data:{}",orders.size());
        try {
     /*       for(int i = 0;i < orders.size();i++) {
                JSONObject jsonObject1 = orders.getJSONObject(i);
                threadPoolUtils.execute(()->{
                    kafkaUtils.sendMessage("batch_insert_order_oms",jsonObject1.toJSONString());
                    log.info("insertOrderOms推送成功!");
                });
            }*/
            orders.forEach(x-> threadPoolUtils.execute(()->{
                JSONObject params = (JSONObject) x;
                kafkaUtils.sendMessage("batch_insert_order_oms",params.toJSONString());
                log.info("insertOrderOms推送成功!");
            }));
        }catch(Exception e) {
            log.error("insertOrderOms异常:",e);
        }
    }
}